package ddbs;

import org.json.JSONArray;
import org.json.JSONException;

public class BinWork extends LocalWork implements JobDoneCallback {
    public static final String STATUS = "bwork";
    public LocalWork w1;
    public LocalWork w2;
    String joinBy[] = new String[3];// A.a_id, B.b_id, id
    StringBuffer peerid = new StringBuffer();
    public int isleaf = 0;

    public LocalWork getW1() {
        return this.w1;
    }
    public LocalWork getW2() {
        return this.w2;
    }
    public String getStatus() {
        return BinWork.STATUS;
    }

    /*
     * For two work in one peer
     * 
     * @see ddbs.Work#formal_sql()
     */
    protected String formalSQL() {
        StringBuffer sql = new StringBuffer();
        sql.append("select ");
        DataSet ds1 = w1.getDataSet();
        for (int i = 0; i < ds1.fileds.length; ++i) {
            String filed = ds1.fileds[i];
            // if (!filed.equals(joinBy[0]))
            if (i != 0)
                sql.append(",");
            sql.append(filed);
        }
        DataSet ds2 = w2.getDataSet();
        for (int i = 0; i < ds2.fileds.length; ++i) {
            String filed = ds2.fileds[i];
            if (!sql.toString().contains(filed))
                sql.append("," + filed);
        }
        if (ds1.exfileds != null && ds1.exfileds.length > 0 && this.isleaf == 1) {
            for (int i = 0; i < ds1.exfileds.length; i++) {
                sql.append("," + ds1.exfileds[i]);
            }
        }
        if (ds2.exfileds != null && ds2.exfileds.length > 0 && this.isleaf == 1) {
            for (int i = 0; i < ds2.exfileds.length; i++) {
                if(!ds2.exfileds[i].contains("Customer.id"))
                    sql.append("," + ds2.exfileds[i]);
            }
        }

        sql.append(" from " + ds1.table);
        if (!ds1.table.contains(ds2.table))
            sql.append("," + ds2.table);
        sql.append(" where " + joinBy[0] + "=" + joinBy[1]);
        if (ds1.limit != null && ds1.limit.length > 0) {
            for (int i = 0; i < ds1.limit.length; ++i) {
                sql.append(" and");
                sql.append(" " + ds1.limit[i]);
            }
        }
        if (ds2.limit != null && ds2.limit.length > 0) {
            for (int i = 0; i < ds2.limit.length; ++i) {
                sql.append(" and");
                sql.append(" " + ds2.limit[i]);
            }
        }
        // System.out.println(w1.getStatus()+w2.getStatus());
        if (w1.getStatus() == BinWork.STATUS) {
            sql.append(" and " + ((BinWork) w1).joinBy[0] + "="
                    + ((BinWork) w1).joinBy[1]);
        }
        //System.out.println(sql.toString()+"isleaf="+isleaf);
        return sql.toString();
    }

    public Peer getPeer() {
        if (w1.getPeer() == w2.getPeer())
            return w1.getPeer();
        return null;
    }

    protected void handleLocal() {
        Peer p = getPeer();
        if (p == null) {
            if (!w1.IsJobDone()) {
                w1.donecb = this;
                Sched.assiginWork(w1);
                w1.dealWith();
            }
            if (!w2.IsJobDone()) {
                w2.donecb = this;
                Sched.assiginWork(w2);
                w2.dealWith();
            }
        } else if (p.IsSelf()) {
            queryDatabase();
        } else {
            p.sendWork(this);
        }
    }

    public void handleResponse(MSG msg) {
        this.data = Data.deserialize(msg.len, msg.content);
        System.out.println("BinWork #" + this.id
                + " handleResponse and get data done size="
                + this.data.getSize());
        done();
    }

    private void clearFileds() {
    }

    private void join() {
        String tempjo1 = joinBy[0].substring(joinBy[0].indexOf(".") + 1);
        String tempjo2 = joinBy[1].substring(joinBy[1].indexOf(".") + 1);

        int anum, bnum;
        if ((anum = findfname(w1, tempjo1)) == -1) {
            System.out.println("join error " + tempjo1);
        }
        if ((bnum = findfname(w2, tempjo2)) == -1) {
            System.out.println("join error2 " + tempjo2);
        }
        long timeStart = System.currentTimeMillis();
        //sort(w1.data, anum);
        //sort(w2.data, bnum);
        sort2(w1.data, anum);
        sort2(w2.data, bnum);
        long timeEnd = System.currentTimeMillis();
        long timeval = timeEnd - timeStart;
        System.out.println(">>>>>>>>>>>>>>Sort time :" + timeval);
        merge(anum, bnum);
        clearFileds();
    }

    private int findfname(LocalWork w, String fname) {
        for (int i = 0; i != w.data.filednames.length; i++) {
            if (w.data.filednames[i].contains(fname))
                return i;
        }
        System.out.println("ERROR: " + fname + " ");
        for (int i = 0; i != w.data.filednames.length; i++) {
            System.out.print(w.data.filednames[i] + " ");
        }
        System.out.println("\n\n");
        return -1;
    }

    private Boolean isSorted(JSONArray arr, int num) {
        int last = -1;
        for (int i = 0; i < arr.length(); i++) {
            try {
                JSONArray row = arr.getJSONArray(i);
                int cur = row.getInt(num);
                if (last > cur)
                    return false;
                last = cur;
            } catch (JSONException e) {
                e.printStackTrace();
                return false;
            }
        }
        return true;
    }

    private void sort(Data dt, int num) {
        if (isSorted(dt.jarray, num))
            return;
        JSONArray js[] = new JSONArray[dt.jarray.length()];
        JSONArray temp;
        for (int i = 0; i != js.length; i++) {
            try {
                js[i] = dt.jarray.getJSONArray(0);
                dt.jarray.remove(0);
            } catch (JSONException e) {
                e.printStackTrace();
            }
        }
        for (int i = 0; i != js.length - 1; i++) {
            for (int j = 0; j != js.length - i - 1; j++) {
                try {
                    if (js[j].getInt(num) > js[j + 1].getInt(num)) {
                        temp = js[j];
                        js[j] = js[j + 1];
                        js[j + 1] = temp;
                    }
                } catch (JSONException e) {
                    e.printStackTrace();
                }
            }
        }
        for (int i = 0; i != js.length; i++) {
            dt.jarray.put(js[i]);
        }
    }

    class Element {
        int id;
        JSONArray row;
        Element (JSONArray arr, int num) throws JSONException {
            this.row = arr;
            this.id = arr.getInt(num);
        }
    }

    private static void QuickSort(Element[] pData, int left, int right) {
        int i, j;
        Element middle, temp;
        i = left;
        j = right;
        middle = pData[left];
        while (true) {
            while ((++i) < right - 1 && pData[i].id < middle.id)
                ;
            while ((--j) > left && pData[j].id > middle.id)
                ;
            if (i >= j)
                break;
            temp = pData[i];
            pData[i] = pData[j];
            pData[j] = temp;
        }

        pData[left] = pData[j];
        pData[j] = middle;

        if (left < j)
            QuickSort(pData, left, j);

        if (right > i)
            QuickSort(pData, i, right);
    }

    private void sort2(Data dt, int num) {
        if (isSorted(dt.jarray, num))
            return;
        JSONArray arr = dt.jarray;
        int length = arr.length();
        Element els[] = new Element[length];
        for (int i = 0; i != length; i++) {
            try {
                els[i] = new Element(arr.getJSONArray(i), num);
            } catch (JSONException e) {
                e.printStackTrace();
            }
        }
        QuickSort(els, 0, length);
        dt.jarray = new JSONArray();
        for (int i = 0; i != length; i++) {
            dt.jarray.put(els[i].row);
        }
    }

    private void merge(int anum, int bnum) {
        
        this.data = new Data();
        this.data.filednames = new String[w1.data.filednames.length
                + w2.data.filednames.length];
        for (int i = 0; i != w1.data.filednames.length; i++) {
            this.data.filednames[i] = w1.data.filednames[i];
        }
        for (int i = 0; i != w2.data.filednames.length; i++) {
            this.data.filednames[i + w1.data.filednames.length] = w2.data.filednames[i];
        }
        
        int realnum = 0;
        String realfileds = imsql.substring(imsql.indexOf("select") + 6,
                imsql.indexOf("from")).trim();
        if (realfileds.equals("*")) {
            StringBuffer rf = new StringBuffer();
            for (int i = 0; i != this.ds.fileds.length; i++) {
                if (i != 0)
                    rf.append(",");
                rf.append(this.ds.fileds[i]);
            }
            realfileds = rf.toString();
        }
        String realfiledsnames[];
        int realfiledsnum[];

        int isCusjoin = 0;
        if (ds.table.equals("Customer"))
            isCusjoin = 1;
        int cusId = 0;

        if (isCusjoin == 1) {
            for (int i = 0; i != this.data.filednames.length; i++) {
                if (realfileds.contains(this.data.filednames[i])) {
                    realnum++;
                    if (this.data.filednames[i].contains("id")) {
                        if (cusId == 0) {
                            cusId++;
                        } else {
                            realnum--;
                            this.data.filednames[i] = null;
                        }
                    }
                } else {
                    this.data.filednames[i] = null;
                }
            }
        } else {
            for (int i = 0; i != this.data.filednames.length; i++) {
                if (realfileds.contains(this.data.filednames[i])) {
                    realnum++;
                } else {
                    this.data.filednames[i] = null;
                }
            }

        }
        realfiledsnames = new String[realnum];
        realfiledsnum = new int[realnum];
        realnum = 0;
        for (int i = 0; i != this.data.filednames.length; i++) {
            if (this.data.filednames[i] != null
                    && this.data.filednames[i].length() != 0) {
                realfiledsnames[realnum] = this.data.filednames[i];
                realfiledsnum[realnum] = i;
                realnum++;
            }
        }
        this.data.filednames = new String[realnum];
        this.data.filednames = realfiledsnames;
 
        
        int length1 = this.w1.data.jarray.length();
        int length2 = this.w2.data.jarray.length();
        int length=0;
        
        for (int i=0;i!=realnum;i++){
            if (realfiledsnum[i]+1>this.w1.data.filednames.length){
                length=i;
                break;
            }
            length=i+1;
        }

        //System.out.println("anum="+anum+" bnum="+bnum);
        //System.out.println("w1length="+w1.data.filednames.length+" w2length="+w2.data.filednames.length);
        
        
        int itw1 = 0, itw2 = 0, iitw1 = 0, iitw2 = 0;
        JSONArray arr1 = w1.data.jarray;
        JSONArray arr2 = w2.data.jarray;
        while (itw1 != length1 && itw2 != length2) {
            try {
                JSONArray j1 = arr1.getJSONArray(itw1);
                JSONArray j2 = arr2.getJSONArray(itw2);
               // System.out.println("j1="+j1.toString()+" j2="+j2.toString());
                int d1 = j1.getInt(anum);
                int d2 = j2.getInt(bnum);
                // System.out.println("itw1="+itw1+" itw2="+itw2+" j1="+j1+" j2="+j2);
                if (d1 == d2) {
                    iitw1 = itw1 + 1;
                    while (iitw1 < length1
                            && arr1.getJSONArray(iitw1).getInt(anum) == d1) {
                        iitw1++;
                    }
                    iitw2 = itw2 + 1;
                    while (iitw2 < length2
                            && arr2.getJSONArray(iitw2).getInt(bnum) == d2) {
                        iitw2++;
                    }
                    // System.out.println("anum="+anum+" bnum="+bnum+" itw1="+itw1+","+iitw1+" itw2="+itw2+","+iitw2);
                    for (int i = itw1; i != iitw1; i++) {
                        for (int j = itw2; j != iitw2; j++) {
                            JSONArray newarr = new JSONArray();
                            JSONArray ja = arr1.getJSONArray(i);
                            JSONArray jb = arr2.getJSONArray(j);
                            //System.out.println("ja="+ja.toString()+" jb="+jb.toString());
                            int alength = ja.length();
                            for (int k = 0; k != length; k++) {
                                newarr.put(ja.get(realfiledsnum[k]));
                            }
                            //System.out.println("length="+length+"realnum[k]="+ realfiledsnum[length]+"alength ="+alength+" realnum="+realnum);
                            for (int k = length; k != realnum; k++) {
                                newarr.put(jb.get(realfiledsnum[k]-alength));
                            }
                            this.data.jarray.put(newarr);
                        }
                    }
                    itw1 = iitw1;
                    itw2 = iitw2;
                } else if (j1.getInt(anum) > j2.getInt(bnum)) {
                    itw2++;
                } else {
                    itw1++;
                }
            } catch (JSONException e) {
                e.printStackTrace();
                return;
            }
        }
    }

    @Override
    public void jobdonecb(Object obj) {
        //System.out.println("Work [" + obj.toString() + "] done");
        if (w1.IsJobDone() && w2.IsJobDone()) {
            if (data == null) {
                System.out.println("Work children done and start join "
                        + this.toString());
                //System.out.println("w1="+w1.toString());
                //System.out.println("w2="+w2.toString());
                join();
            }
            done();
        }
    }

    public MSG toMSG() {
        getPeerid(w1);
        getPeerid(w2);
        String pid = peerid.toString();
        int peerlength = pid.length();
        String content = this.imsql + pid + peerlength;
        peerid = new StringBuffer();
        ;
        int len = content.length();
        MSG SWMsg = new MSG(Sched.msgType, BinWork.STATUS, this.id, len,
                content);
        return SWMsg;
    }

    public void getPeerid(LocalWork w) {
        // System.out.println(w.getStatus());
        if (w.getStatus() == Work.STATUS) {
            peerid.append(w.ds.peer.getId());
        } else {
            getPeerid(((BinWork) w).w1);
            getPeerid(((BinWork) w).w2);
        }
        return;
    }
}
